package cn.com.itg.function.watetmarks;

import cn.com.itg.pojo.domain.vo.MatdocSendVO;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;

import javax.annotation.Nullable;

public class MatdocSendAPW implements AssignerWithPeriodicWatermarks<MatdocSendVO>{
    private long currentMaxTs = Long.MIN_VALUE;

    @Nullable
    @Override
    public Watermark getCurrentWatermark() {
        return new Watermark(currentMaxTs);
    }

    @Override
    public long extractTimestamp(MatdocSendVO record, long l) {
        //获取当前记录的时间戳
        long recordTimeStamp = record.getTimestamp();
        //更新最大时间戳
        currentMaxTs = Math.max(recordTimeStamp, currentMaxTs);
        //返回当前记录时间戳
        return recordTimeStamp;
    }
}
